package rabbitmq

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"math/rand"
	"strings"
	"sync"
	"time"
)

type RabbitMq struct {
	AmqpUrl       string //amqp地址
	ConnectionNum int    //连接数
	ChannelNum    int    //每个连接的channel数量

	connections    map[int]*amqp.Connection
	channels       map[int]channel
	idelChannels   []int
	busyChannels   map[int]int
	m              *sync.Mutex
	wg             *sync.WaitGroup
	ctx            context.Context
	cancel         context.CancelFunc
	connectIdChan  chan int
	lockConnectIds map[int]bool
}

type channel struct {
	ch            *amqp.Channel
	notifyClose   chan *amqp.Error
	notifyConfirm chan amqp.Confirmation
}

const (
	retryCount      = 5
	waitConfirmTime = 3 * time.Second
)

func (r *RabbitMq) Run(closeSig chan bool) {
	r.InitAmqp()

	log.Println("InitAmqp start successful")
	<-closeSig

	for _, chanal := range r.channels {
		chanal.ch.Close()
	}
	log.Println("channels close successful")

	for _, connection := range r.connections {
		connection.Close()
	}
	log.Println("connections close successful")
}

func (r *RabbitMq) OnDestroy() {
	r.InitAmqp()
}

func (r *RabbitMq) InitAmqp() {
	if r.AmqpUrl == "" {
		log.Fatal("rabbitmq's address can not be empty!")
	}
	if r.ConnectionNum == 0 {
		r.ConnectionNum = 10
	}
	if r.ChannelNum == 0 {
		r.ChannelNum = 10
	}
	r.m = new(sync.Mutex)
	r.wg = new(sync.WaitGroup)
	r.ctx, r.cancel = context.WithTimeout(context.Background(), waitConfirmTime)
	r.lockConnectIds = make(map[int]bool)
	r.connectIdChan = make(chan int)

	r.connectPool()
	r.channelPool()

	for cnt := 0; cnt < 10; cnt++ {
		r.PutIntoQueue("ys.exchange", "ys.msg.normal", fmt.Sprint("msg-normal", cnt))
		r.PutIntoQueue("ys.exchange", "ys.msg.vip", fmt.Sprint("msg-vip", cnt))
	}
}

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

func (r *RabbitMq) connectPool() {
	r.connections = make(map[int]*amqp.Connection)
	for i := 0; i < r.ConnectionNum; i++ {
		connection := r.connect()
		r.connections[i] = connection
	}
}

func (r *RabbitMq) channelPool() {
	r.channels = make(map[int]channel)
	//r.idelChannels = make(map[int]int)
	for index, _ := range r.connections {
		for j := 0; j < r.ChannelNum; j++ {
			key := index*r.ChannelNum + j
			r.channels[key] = r.createChannel(index)
			//r.idelChannels[key] = key
			r.idelChannels = append(r.idelChannels, key)
		}
	}
}

func (r *RabbitMq) connect() *amqp.Connection {
	conn, err := amqp.Dial(r.AmqpUrl)
	failOnError(err, "Failed to connect to RabbitMQ")
	//defer conn.Close()
	return conn
}

func (r *RabbitMq) recreateChannel(connectId int, err error) (ch *amqp.Channel) {
	if strings.Index(err.Error(), "channel/connection is not open") >= 0 || strings.Index(err.Error(), "CHANNEL_ERROR - expected 'channel.open'") >= 0 {
		//r.connections[connectId].Close()
		if r.connections[connectId].IsClosed() {
			r.lockWriteConnect(connectId)
		}
		ch, err = r.connections[connectId].Channel()
		failOnError(err, "Failed to open a channel")
	} else {
		failOnError(err, "Failed to open a channel")
	}
	return
}

func (r *RabbitMq) lockWriteConnect(connectId int) {

	r.m.Lock()
	if !r.lockConnectIds[connectId] {
		r.lockConnectIds[connectId] = true
		r.m.Unlock()

		go func(connectId int) {
			r.wg.Add(1)
			defer r.wg.Done()

			r.connections[connectId] = r.connect()
			r.connectIdChan <- connectId

		}(connectId)
	} else {
		r.m.Unlock()
	}

	for {
		select {
		case cid := <-r.connectIdChan:

			delete(r.lockConnectIds, cid)

			if len(r.lockConnectIds) == 0 {
				r.wg.Wait()
				return
			} else {
				continue
			}
		case <-time.After(waitConfirmTime):
			r.lockConnectIds = make(map[int]bool)
			r.wg.Wait()
			return
		}
	}
}

func (r *RabbitMq) createChannel(connectId int) channel {
	var notifyClose = make(chan *amqp.Error)
	var notifyConfirm = make(chan amqp.Confirmation)

	cha := channel{
		notifyClose:   notifyClose,
		notifyConfirm: notifyConfirm,
	}
	if r.connections[connectId].IsClosed() {
		r.lockWriteConnect(connectId)
	}
	ch, err := r.connections[connectId].Channel()
	if err != nil {
		ch = r.recreateChannel(connectId, err)
	}

	ch.Confirm(false)
	ch.NotifyClose(cha.notifyClose)
	ch.NotifyPublish(cha.notifyConfirm)

	cha.ch = ch
	//go func() {
	//	select {
	//	case <-cha.notifyClose:
	//		fmt.Println("close channel")
	//	}
	//}()
	return cha
}

func (r *RabbitMq) getChannel() (*amqp.Channel, int) {
	r.m.Lock()
	defer r.m.Unlock()
	idelLength := len(r.idelChannels)
	if idelLength > 0 {

		rand.Seed(time.Now().Unix())
		index := rand.Intn(idelLength)
		channelId := r.idelChannels[index]
		r.idelChannels = append(r.idelChannels[:index], r.idelChannels[index+1:]...)
		r.busyChannels = make(map[int]int)
		r.busyChannels[channelId] = channelId

		ch := r.channels[channelId].ch
		//fmt.Println("channels count: ",len(r.channels))
		//fmt.Println("idel channels count: ",len(r.idelChannels))
		//fmt.Println("busy channels count: ",len(r.busyChannels))
		//fmt.Println("channel id: ",channelId)
		return ch, channelId
	} else {
		//return r.createChannel(0,r.connections[0]),-1
		return nil, -1
	}
}

func (r *RabbitMq) declareExchange(ch *amqp.Channel, exchangeName string, channelId int) *amqp.Channel {
	err := ch.ExchangeDeclare(
		exchangeName, // name
		"topic",      // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // no-wait
		nil,          // arguments
	)
	if err != nil {
		ch = r.reDeclareExchange(channelId, exchangeName, err)
	}
	return ch
}

func (r *RabbitMq) reDeclareExchange(channelId int, exchangeName string, err error) (ch *amqp.Channel) {
	//fmt.Println("reDeclareExchange")

	var connectionId int
	if strings.Index(err.Error(), "channel/connection is not open") >= 0 {

		//r.channels[channelId].Close()
		if channelId == -1 {
			rand.Seed(time.Now().Unix())
			index := rand.Intn(r.ConnectionNum)
			connectionId = index
		} else {
			connectionId = int(channelId / r.ChannelNum)
		}
		cha := r.createChannel(connectionId)

		r.lockWriteChannel(channelId, cha)
		//r.channels[channelId] = cha
		err := cha.ch.ExchangeDeclare(
			exchangeName, // name
			"topic",      // type
			true,         // durable
			false,        // auto-deleted
			false,        // internal
			false,        // no-wait
			nil,          // arguments
		)
		if err != nil {
			failOnError(err, "Failed to declare an exchange")
		}
		return cha.ch
	} else {

		failOnError(err, "Failed to declare an exchange")
		return nil
	}
}

func (r *RabbitMq) lockWriteChannel(channelId int, cha channel) {
	r.m.Lock()
	defer r.m.Unlock()
	r.channels[channelId] = cha
}

func (r *RabbitMq) dataForm(notice interface{}) string {
	body, err := json.Marshal(notice)
	if err != nil {
		log.Panic(err)
	}
	return string(body)
}

func (r *RabbitMq) publish(channelId int, ch *amqp.Channel, exchangeName string, routeKey string, data string) (err error) {

	err = ch.Publish(
		exchangeName, // exchange
		routeKey,     //severityFrom(or.Args), // routing key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         []byte(data),
		})

	if err != nil {
		if strings.Index(err.Error(), "channel/connection is not open") >= 0 {
			err = r.rePublish(channelId, exchangeName, err, routeKey, data)
		}
	}
	return
}

func (r *RabbitMq) rePublish(channelId int, exchangeName string, errmsg error, routeKey string, data string) (err error) {
	//fmt.Println("rePublish")

	ch := r.reDeclareExchange(channelId, exchangeName, errmsg)
	err = ch.Publish(
		exchangeName, // exchange
		routeKey,     //severityFrom(or.Args), // routing key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         []byte(data),
		})
	return
}

func (r *RabbitMq) backChannelId(channelId int, ch *amqp.Channel) {
	r.m.Lock()
	defer r.m.Unlock()
	r.idelChannels = append(r.idelChannels, channelId)
	delete(r.busyChannels, channelId)
	return
}

func (r *RabbitMq) PutIntoQueue(exchangeName string, routeKey string, notice interface{}) (message interface{}, puberr error) {
	defer func() {
		msg := recover()
		if msg != nil {
			//fmt.Println("msg: ",msg)
			puberrMsg, _ := msg.(string)
			//fmt.Println("ok: ",ok)
			//fmt.Println("puberrMsg : ",puberrMsg)
			puberr = errors.New(puberrMsg)
			return
		}
	}()

	ch, channelId := r.getChannel()
	if ch == nil {
		rand.Seed(time.Now().Unix())
		index := rand.Intn(r.ConnectionNum)
		cha := r.createChannel(index)
		defer cha.ch.Close()
		ch = cha.ch
		//fmt.Println("ch: ",ch)
	}
	ch = r.declareExchange(ch, exchangeName, channelId)

	data := r.dataForm(notice)
	var tryTime = 1

	for {
		puberr = r.publish(channelId, ch, exchangeName, routeKey, data)
		if puberr != nil {
			if tryTime <= retryCount {
				//log.Printf("%s: %s", "Failed to publish a message, try again.", puberr)
				tryTime++
				continue
			} else {
				//log.Printf("%s: %s data: %s", "Failed to publish a message", puberr,data)
				r.backChannelId(channelId, ch)
				return notice, puberr
			}
		}

		select {
		case confirm := <-r.channels[channelId].notifyConfirm:
			if confirm.Ack {
				//log.Printf(" [%s] Sent %d message %s", routeKey, confirm.DeliveryTag, data)
				r.backChannelId(channelId, ch)

				return notice, nil
			}
		case <-time.After(waitConfirmTime):
			//	log.Printf("message: %s data: %s", "Can not receive the confirm.", data)
			r.backChannelId(channelId, ch)
			confirmErr := errors.New("Can not receive the confirm . ")
			return notice, confirmErr
		}
	}
}

/*
func (r *RabbitMq) GetFromQueue(queue string, exchangeName string, routeKey string) (message interface{}, puberr error) {
	defer func() {
		msg := recover()
		if msg != nil {
			//fmt.Println("msg: ",msg)
			puberrMsg, _ := msg.(string)
			//fmt.Println("ok: ",ok)
			//fmt.Println("puberrMsg : ",puberrMsg)
			puberr = errors.New(puberrMsg)
			return
		}
	}()

	ch, channelId := r.getChannel()
	if ch == nil {
		rand.Seed(time.Now().Unix())
		index := rand.Intn(r.ConnectionNum)
		cha := r.createChannel(index)
		defer cha.ch.Close()
		ch = cha.ch
		//fmt.Println("ch: ",ch)
	}
	ch = r.declareExchange(ch, exchangeName, channelId)

	//data := r.dataForm(notice)
	var tryTime = 1

	for {
		puberr = r.publish(channelId, ch, exchangeName, routeKey, data)
		if puberr != nil {
			if tryTime <= retryCount {
				//log.Printf("%s: %s", "Failed to publish a message, try again.", puberr)
				tryTime++
				continue
			} else {
				//log.Printf("%s: %s data: %s", "Failed to publish a message", puberr,data)
				r.backChannelId(channelId, ch)
				return notice, puberr
			}
		}

		select {
		case confirm := <-r.channels[channelId].notifyConfirm:
			if confirm.Ack {
				//log.Printf(" [%s] Sent %d message %s", routeKey, confirm.DeliveryTag, data)
				r.backChannelId(channelId, ch)

				return notice, nil
			}
		case <-time.After(waitConfirmTime):
			//	log.Printf("message: %s data: %s", "Can not receive the confirm.", data)
			r.backChannelId(channelId, ch)
			confirmErr := errors.New("Can not receive the confirm . ")
			return notice, confirmErr
		}
	}
}
*/
